package com.wistech.solr.ik;

import java.util.Collection;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPubSub;

public class RedisIKWordProvider extends IKWordProvider {
	private JedisPool jedisPool;
	private String channel;
	private String key;

	public RedisIKWordProvider(JedisPool jedisPool, String channel) {
		this.jedisPool = jedisPool;
		this.channel = channel;
		this.key = RedisIKWordProvider.class.getName();
		this.subscribe();
	}

	@Override
	public void loadWords() {
		Jedis jedis = jedisPool.getResource();
		Collection<String> words = jedis.smembers(key);
		addWordToIKDictionary(words);
		jedisPool.returnResource(jedis);
	}

	@Override
	protected void saveWord(final String word) {
		Jedis jedis = jedisPool.getResource();
		jedis.sadd(key, word);
		jedisPool.returnResource(jedis);
	}

	@Override
	public void removeWord(String word) {
		Jedis jedis = jedisPool.getResource();
		jedis.srem(key, word);
		jedisPool.returnResource(jedis);
	}

	@Override
	public void cleanWords() {
		Jedis jedis = jedisPool.getResource();
		jedis.del(key);
		jedisPool.returnResource(jedis);
	}

	@Override
	protected void publishWord(String word) {
		publish(word);
	}

	public void publish(final String word) {
		new Thread(new Runnable() {
			@Override
			public void run() {
				Jedis jedis = jedisPool.getResource();
				jedis.publish(channel, word);
				jedisPool.returnResource(jedis);
			}
		}).start();
	}

	private void subscribe() {
		new Thread(new Runnable() {
			@Override
			public void run() {
				Jedis jedis = jedisPool.getResource();
				jedis.subscribe(new NewsListener(), channel);
				jedisPool.returnResource(jedis);
				System.out.println("subscribed!!!!!!!!!");
			}
		}).start();
	}

	private class NewsListener extends JedisPubSub {
		@Override
		public void onMessage(String channel, String message) {
			System.out.println("receive message from redis: " + message);
			addWordToIKDictionary(message);
		}

		@Override
		public void onPMessage(String pattern, String channel, String message) {
		}

		@Override
		public void onSubscribe(String channel, int subscribedChannels) {
		}

		@Override
		public void onUnsubscribe(String channel, int subscribedChannels) {
		}

		@Override
		public void onPUnsubscribe(String pattern, int subscribedChannels) {
		}

		@Override
		public void onPSubscribe(String pattern, int subscribedChannels) {
		}
	}

}
